package day1

import org.apache.log4j.{Level, Logger}
import org.apache.spark.storage.StorageLevel
import org.apache.spark.{SparkConf, SparkContext}

object RDDStorage {
  def main(args: Array[String]): Unit = {
    System.setProperty("hadoop.home.dir", "D:\\新建文件夹\\HCIA初级\\hadoop")
    Logger.getLogger("org").setLevel(Level.OFF)
    val sc = new SparkContext(new SparkConf().setAppName("Basic").setMaster("local"))
    //持久化

    //当一个RDD转化为其他RDD并使用行动算子的完成之后，所以的内容都会被释放掉
    val arr1 = Array(1,2,3,4,5,6,7,8,9)
    val rdd1 = sc.parallelize(arr1)
    val rdd2 = rdd1.map(i => i)
    rdd2.foreach(println)
    rdd2.foreach(println)

    println("******************************************")
    //在遇到连接操作、需要网络进行大量数据传递、计算过程复杂
    //需要考虑对于部分RDD进行持久化操作：persist()
    /*
      val NONE = new StorageLevel(false, false, false, false)
      val DISK_ONLY = new StorageLevel(true, false, false, false)
      val DISK_ONLY_2 = new StorageLevel(true, false, false, false, 2)
      val DISK_ONLY_3 = new StorageLevel(true, false, false, false, 3)
      val MEMORY_ONLY = new StorageLevel(false, true, false, true)
      val MEMORY_ONLY_2 = new StorageLevel(false, true, false, true, 2)
      val MEMORY_ONLY_SER = new StorageLevel(false, true, false, false)
      val MEMORY_ONLY_SER_2 = new StorageLevel(false, true, false, false, 2)
      val MEMORY_AND_DISK = new StorageLevel(true, true, false, true)
      val MEMORY_AND_DISK_2 = new StorageLevel(true, true, false, true, 2)
      val MEMORY_AND_DISK_SER = new StorageLevel(true, true, false, false)
      val MEMORY_AND_DISK_SER_2 = new StorageLevel(true, true, false, false, 2)
      val OFF_HEAP = new StorageLevel(true, true, true, false, 1)
     */
    /*
     * Spark运行处理过程的持久化
     * 所以RDD的默认持久化类型：NONE，表示不持久
     * DISK、MEMORY、SER、num
     * DISK：持久化到硬盘
     * MEMORY：持久化到内存，如果出现内存溢出，抛出异常
     * SER：序列化，将数据进行压缩存放在内存
     * num：数字，副本个数
     */
    rdd2.persist(StorageLevel.DISK_ONLY_3)
    rdd2.foreach(println)

    println("******************************************")

    /*
     * 在rdd2中，如果做了持久化操作，那么此时rdd3直接从内存中读取rdd2的内容进行计算
     * 如果rdd2没有实现持久化操作，那么此时rdd3就会重新从rdd1-->rdd2-->rdd3
     */
    val rdd3 = rdd2.map(_ * 3)
    rdd3.foreach(println)

  }

}